package pl.edu.agh.broker.proxy;

import java.io.InputStream;
import java.util.Properties;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.network.jms.JmsTopicConnector;

import pl.edu.agh.broker.mns.client.DefaultResponder;
import pl.edu.agh.broker.mns.client.MultiNameServiceLocator;
import pl.edu.agh.broker.mns.shared.MultiNameService;
import pl.edu.agh.stock.data.IndexDataPoint;

public class ProxyBroker {

	private static ProxyBroker instance = new ProxyBroker();

	private static final String PROPERTY_FILE = "broker.properties";
	private static final String INFO = " INFO ";

	private static final String BROKER_URL = "broker.url";
	private static final String BROKER_HTTP_URL = "http.port";
	private static final String BROKER_ID = "broker.id";
	private static final String BROKER_TOPIC = "broker.topic";
	private static final String MNS_ADDRESS = "mns.address";
	private static final String MNS_PORT = "mns.port";
	private Properties properties;

	private IndexDataPoint indexDataPoint;
	private String path;

	private ProxyBroker() {

	}

	public static ProxyBroker getInstance() {
		return instance;
	}

	public void init(String path) {
		this.path = path;
		try {
			this.startProxyBroker();
			this.startConsumer();
		} catch (Exception e) {
			e.printStackTrace();
		}
		System.out.println(INFO + "Proxy Broker started");
	}

	public IndexDataPoint getIndex() {
		return indexDataPoint;
	}

	private void startProxyBroker() throws Exception {

		InputStream inStream = Thread.currentThread().getContextClassLoader()
				.getResourceAsStream(PROPERTY_FILE);

		// load properties
		properties = new Properties();
		properties.load(inStream);

		// resolve main broker address
		MultiNameService mns = MultiNameServiceLocator.getMultiNameService(
				properties.getProperty(MNS_ADDRESS), Integer
						.parseInt(properties.getProperty(MNS_PORT)));

		String remoteBrokerURL = (String) mns.getAny(properties
				.getProperty(BROKER_ID));

		// create topic factory
		ActiveMQConnectionFactory foreignTopicConnectionFactory = new ActiveMQConnectionFactory(
				remoteBrokerURL);

		// configure and start broker
		BrokerService brokerService = BrokerFactory
				.createBroker("xbean:activemq.xml");
		JmsTopicConnector jmsTopicConnector = (JmsTopicConnector) brokerService
				.getJmsBridgeConnectors()[0];
		jmsTopicConnector
				.setOutboundTopicConnectionFactory(foreignTopicConnectionFactory);
		brokerService.start();

		// register broker
		mns.add(properties.getProperty(BROKER_TOPIC), properties
				.getProperty(BROKER_HTTP_URL)
				+ path + "/rest/index", new DefaultResponder());
	}

	private void startConsumer() throws JMSException {

		String brokerURL = properties.getProperty(BROKER_URL);

		ActiveMQConnectionFactory localConnectionFactory = new ActiveMQConnectionFactory(
				brokerURL);
		TopicConnection connection = localConnectionFactory
				.createTopicConnection();
		connection.start();
		TopicSession session = connection.createTopicSession(false,
				Session.AUTO_ACKNOWLEDGE);
		Topic topic = session.createTopic(properties.getProperty(BROKER_TOPIC));
		TopicSubscriber topicSubscriber = session.createSubscriber(topic);

		topicSubscriber.setMessageListener(new MessageListener() {
			@Override
			public void onMessage(Message message) {
				try {
					if (message instanceof ObjectMessage) {
						ObjectMessage objectMessage = (ObjectMessage) message;
						IndexDataPoint stockDataPoint = (IndexDataPoint) objectMessage
								.getObject();
						if (stockDataPoint != null) {
							indexDataPoint = stockDataPoint;
						}
						System.out.println(INFO + "Received index: "
								+ stockDataPoint.getName() + " value "
								+ stockDataPoint.getValors());
					} else {
						System.out.println(INFO + "Invalid message received.");
					}
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		});
	}

}
